package com.bigdata.warehouse.dws;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @BelongsProject: realtime-project-10zlq
 * @BelongsPackage: com.bigdata.warehouse.dws
 * @Author: zhangleqing
 * @CreateTime: 2025-08-22  20:07
 * @Description:
 * @Version: 1.0
 * 每天车辆的进入和出去
 */
public class DayCarsInOut {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);


        // 1.dwd车辆明细表
        tEnv.executeSql("CREATE TABLE dwd_cars_log (\n" +
                " id STRING,\n" +
                " owerId INT,\n" +
                " opTime STRING,\n" +
                " cId BIGINT,\n" +
                " carCode STRING,\n" +
                " ctype SMALLINT,\n" +
                " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(opTime),0),\n" +
                " WATERMARK FOR time_ltz AS time_ltz - INTERVAL '1' SECOND"+
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'dwd_cars_log',\n" +
                " 'properties.bootstrap.servers' = 'node101:9092',\n" +
                " 'properties.group.id' = 'dwd_cars_log',\n" +
                " 'scan.startup.mode' = 'earliest-offset',\n" +
                " 'format' = 'json'\n" +
                ")\n");


        // 2.读取用户维表 dim_base_ower
        tEnv.executeSql(
                "CREATE TABLE dim_base_ower (\n" +
                        "    id INT,\n" +
                        "    roomId INT,\n" +
                        "    owerCode STRING,\n" +
                        "    owerName STRING,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerIdcard STRING,\n" +
                        "    owerTel STRING,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    remark STRING,\n" +
                        "    createTime STRING,\n" +
                        "    operatorTime STRING\n" +
                        ") WITH (\n" +
                        "'connector' = 'doris',\n" +
                        "'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "'table.identifier' = 'sca_doris.ower',\n" +
                        "'username' = 'root',\n" +
                        "'password' = '123456'\n" +
                        ")"
        );


        // 3.读取车辆维度表 dim_base_cars
        tEnv.executeSql(
                "CREATE TABLE dim_base_cars (\n" +
                        "    id INT,\n" +
                        "    owerId INT,\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    `type` STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.cars',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );


        // 4.读取社区维度表 dim_base_community
        tEnv.executeSql(
                "CREATE TABLE dim_base_community (\n" +
                        "    id INT,\n" +
                        "    communityName STRING,\n" +
                        "    district_id INT,\n" +
                        "    city_id INT,\n" +
                        "    communityLocation STRING,\n" +
                        "    remark STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.community',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );


        // 5.读取地区维度表 dim_base_district
        tEnv.executeSql(
                "CREATE TABLE dim_base_district (\n" +
                        "    id INT,\n" +
                        "    pid INT,\n" +
                        "    code STRING,\n" +
                        "    `name` STRING,\n" +
                        "    remark STRING,\n" +
                        "    create_time STRING,\n" +
                        "    update_time STRING,\n" +
                        "    status STRING,\n" +
                        "    `level` STRING\n" +
                        ") WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030', -- 通常需要端口\n" +
                        "    'table.identifier' = 'sca_doris.district',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456'\n" +
                        ")"
        );


        // 6.窗口计算
        Table windowTable = tEnv.sqlQuery(
                "select  "+
                        "TUMBLE_START(time_ltz,INTERVAL '1' second) as stt,"+
                        "TUMBLE_END(time_ltz,INTERVAL '1' second) as edt,"+
                        "CURRENT_TIMESTAMP as ts,"+
                        "cId,"+
                        "owerId,"+
                        "carCode,"+
                        "sum(if(ctype=1,1,0)) car_in_cnt,"+
                        "sum(if(ctype=2,1,0)) car_out_cnt "+
                        " from dwd_cars_log GROUP BY TUMBLE(time_ltz,INTERVAL '1' second),cId,owerId,carCode "
        );

        // windowTable.execute().print();
        // 创建临时视图
        tEnv.createTemporaryView("windowTable",windowTable);


/* dws层 每天车辆进出统计 dws_car_behavior
CREATE TABLE dws_car_behavior (
    carCode STRING,
    carColor STRING,
    type STRING,
    owerId INT,
    owerSex INT,
    owerAge INT,
    owerHomeplace STRING,
    owerWorkplace STRING,
    community_id INT,
    communityName STRING,
    city_id INT,
    district_id INT,
    district_name STRING,
    car_in_cnt INT,
    car_out_cnt INT,
    stt TIMESTAMP(3),
    edt TIMESTAMP(3),
    ts TIMESTAMP_LTZ(3),
    PRIMARY KEY (carCode, stt, edt) NOT ENFORCED
) WITH (
'connector' = 'upsert-kafka',
'topic' = 'dws_car_behavior',
'properties.bootstrap.servers' = 'node101:9092',
'key.format' = 'json',
'value.format' = 'json'
)
 */
        // 每天车辆进出表 dws_car_behavior
        tEnv.executeSql(
                "CREATE TABLE dws_car_behavior (\n" +
                        "    carCode STRING,\n" +
                        "    carColor STRING,\n" +
                        "    type STRING,\n" +
                        "    owerId INT,\n" +
                        "    owerSex STRING,\n" +
                        "    owerAge INT,\n" +
                        "    owerHomeplace STRING,\n" +
                        "    owerWorkplace STRING,\n" +
                        "    community_id INT,\n" +
                        "    communityName STRING,\n" +
                        "    city_id INT,\n" +
                        "    district_id INT,\n" +
                        "    district_name STRING,\n" +
                        "    car_in_cnt INT,\n" +
                        "    car_out_cnt INT,\n" +
                        "    stt TIMESTAMP(3),\n" +
                        "    edt TIMESTAMP(3),\n" +
                        "    ts TIMESTAMP_LTZ(3),\n" +
                        "    PRIMARY KEY (carCode, stt, edt) NOT ENFORCED\n" +
                        ") WITH (\n" +
                        "'connector' = 'upsert-kafka',\n" +
                        "'topic' = 'dws_car_behavior',\n" +
                        "'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "'key.format' = 'json',\n" +
                        "'value.format' = 'json'\n" +
                        ")"
        );

        // 插入语句
        tEnv.executeSql(
                "insert into dws_car_behavior SELECT \n" +
                        "    car.carCode,\n" +
                        "    car.carColor,\n" +
                        "    car.type,\n" +
                        "    ol.id as owerId,\n" +
                        "    ol.owerSex,\n" +
                        "    ol.owerAge,\n" +
                        "    ol.owerHomeplace,\n" +
                        "    ol.owerWorkplace,\n" +
                        "    commu.id community_id,\n" +
                        "    commu.communityName,\n" +
                        "    commu.city_id,\n" +
                        "    commu.district_id,\n" +
                        "    distri.name district_name,\n" +
                        "    cl.car_in_cnt,\n" +
                        "    cl.car_out_cnt,\n" +
                        "    cl.stt,\n" +
                        "    cl.edt,\n" +
                        "    cl.ts \n" +
                        "FROM windowTable cl \n" +
                        "JOIN dim_base_ower ol ON cl.owerId = ol.id \n" +
                        "JOIN dim_base_cars car ON cl.carCode = car.carCode \n" +
                        "JOIN dim_base_community commu ON cl.cId = commu.id \n" +
                        "JOIN dim_base_district distri ON commu.district_id = distri.id"
        );

        // 查询语句 前10即可
        tEnv.executeSql(
                "select * from dws_car_behavior limit 10"
        ).print();

        // env.execute("DayCarsInOut");
    }
}
